#include "TcpConnection.h"
#include "EventLoop.h"
#include "Socket.h"
#include "Channel.h"
#include "Logger.h"

#include <functional>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <strings.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

TcpConnection::TcpConnection(EventLoop* loop,
              const std::string& name,
              int sockfd,
              const InetAddress& localAddr,
              const InetAddress& peerAddr)
  : m_loop(loop),
    m_name(name),
    m_socket(new Socket(sockfd)),
    m_channel(new Channel(loop, sockfd)),
    m_state(kConnecting),
    m_localAddr(localAddr),
    m_peerAddr(peerAddr),
    m_reading(false),
    m_highWaterMark(64 * 1024 * 1024)
{
  m_channel->setReadCallback(std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
  m_channel->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
  m_channel->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
  m_channel->setErrorCallback(std::bind(&TcpConnection::handleError, this));

  LOG_INFO("TcpConnection::ctor[%s] at fd=%d\n", name.c_str(), sockfd);
}

TcpConnection::~TcpConnection()
{
  LOG_INFO("TcpConnection::dtoc[%s] at fd = %d", m_name.c_str(), m_socket->fd());
}

void TcpConnection::handleRead(Timestamp receiveTime)
{
  int savedErrno = 0;
  ssize_t n = m_inputBuffer.readFd(m_socket->fd(), &savedErrno);
  // 代表有数据了
  if(n > 0) {
    // 已建立链接的用户有可读事件了
    m_messageCallback(shared_from_this(), &m_inputBuffer, receiveTime);
  } else if (n == 0) {
    handleClose();
  } else {
    errno = savedErrno;
    LOG_ERROR("TcpConnection::handleRead\n");
  }
}
void TcpConnection::handleWrite()
{
  LOG_INFO("hanshiyu debug handleWrite");
  int savedErrno = 0;
  if (m_channel->isWriting()) {
    ssize_t n = m_outputBuffer.writeFd(m_socket->fd(), &savedErrno);
    if (n > 0) {
      m_outputBuffer.retrieve(n);
      if (m_outputBuffer.readableBytes() == 0) {
        m_channel->disableWriting();
        if (m_writeCompleteCallback) {
        // 唤醒 loop 对应的线程进行回调
        m_loop->queueInLoop(std::bind(m_writeCompleteCallback, shared_from_this()));
        }
        if (m_state == kDisconnecting) {
          shutdownInLoop();
        }
      }
    } else {
      LOG_ERROR("handle Write error");
    }
  } else {
    LOG_ERROR("TcpConnection fd=%d down, no more writing\n", m_channel->fd());
  } 
}
void TcpConnection::handleClose()
{
  LOG_INFO("TcpConnection::handleClose - %s\n", m_name.c_str());
  setState(kDisconnected);
  m_channel->disableAll();
  TcpConnectionPtr connPtr(shared_from_this());
  m_connectionCallback(connPtr);
  m_closeCallback(connPtr);
}
void TcpConnection::handleError()
{
  LOG_INFO("TcpConnection::handleError - %s\n", m_name.c_str());
  int optval;
  socklen_t option = sizeof(optval);
  int err = 0;
  if (::getsockopt(m_channel->fd(), SOL_SOCKET, SO_ERROR, &optval, &option) < 0) {
    err = errno;
  } else {
    err = optval;
  }
  LOG_ERROR("TcpConnection::handleError - SO_ERROR = %d\n", err);
}

void TcpConnection::send(const void* message, int len)
{
  send(std::string((char*)message, len));
}

void TcpConnection::send(const std::string& message)
{
  if (m_state == kConnected) {
    if (m_loop->isInLoopThread()) {
      sendMessageInLoop(message.c_str(), message.size());
    } else {
      m_loop->runInLoop(std::bind(&TcpConnection::sendMessageInLoop, this, message.c_str(), message.size()));
    }
  }
}

void TcpConnection::sendMessageInLoop(const void* message, int len)
{
  LOG_INFO("hanshiyu debug %s", (char*)message);
  ssize_t nwrote = 0;
  size_t remaining = len;
  bool faultError = false;
  if (m_state == kDisconnected) {
    LOG_ERROR("disconnected, give up writing\n");
    return;
  }
  // 第一次写数据
  if (!m_channel->isWriting() && m_outputBuffer.readableBytes() == 0) {
    nwrote = ::write(m_socket->fd(), message, len);
    if (nwrote >= 0) {
      remaining = len - nwrote;
      if (remaining == 0 && m_writeCompleteCallback) {
        m_loop->queueInLoop(std::bind(m_writeCompleteCallback, shared_from_this()));
      }
    } else {
      nwrote = 0;
      if (errno!= EWOULDBLOCK) {
        if (errno == EPIPE || errno == ECONNRESET) faultError = true;
      }
    }
    // 说明这次 write 没有全部发送出去
    if (!faultError && remaining > 0) {
      size_t oldLen = m_outputBuffer.readableBytes();
      if (oldLen + remaining >= m_highWaterMark && oldLen < m_highWaterMark && m_highWaterMarkCallback) {
        m_loop->queueInLoop(std::bind(m_highWaterMarkCallback, shared_from_this(), oldLen + remaining));
      }
      m_outputBuffer.append((char*)message + nwrote, remaining);
      if (!m_channel->isWriting()) {
        m_channel->enableWriting();
      }
    }
  }
}

void TcpConnection::connectEstablished() 
{
  setState(kConnected);
  m_channel->tie(shared_from_this());
  m_channel->enableReading();
  m_connectionCallback(shared_from_this());
}
void TcpConnection::connectDestroyed()
{
  LOG_INFO("TcpConnection::connectDestroyed - %s\n", m_name.c_str());
  if (m_state == kConnected) {
    setState(kDisconnected);
    m_channel->disableAll();
    TcpConnectionPtr connPtr(shared_from_this());
    m_connectionCallback(connPtr);
    m_closeCallback(connPtr);
  }
  m_channel->remove();
}

void TcpConnection::shutdown()
{
  if (m_state == kConnected) {
    setState(kDisconnecting);
    if (m_loop->isInLoopThread()) {
      shutdownInLoop();
    } else {
      m_loop->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
  }  
}

void TcpConnection::shutdownInLoop()
{
  if (!m_channel->isWriting()) {
    m_socket->shutdownWrite();
  }
}